package streaming
import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object ChangeInfoAndSend {

  def getKafkaConsumerParameters(groupid: String):Map[String,Object] = {
  Map[String,Object](
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG->"linux121:9092,linux122:9092,linux123:9093",
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG->classOf[StringDeserializer],
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG->classOf[StringDeserializer],
    ConsumerConfig.GROUP_ID_CONFIG->groupid,
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG->(false:java.lang.Boolean),
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG->"earliest"
    )
  }

  def sendToMytopic2(msg:String,topic:String):Unit={
    val producer = new KafkaProducer[String, String](getKafkaProducerParameters())
    val record = new ProducerRecord[String, String](topic, msg)
    producer.send(record)
  }


  def getKafkaProducerParameters(): Properties = {
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop
  }

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    val topics: Array[String] = Array("mytopic1")
    val groupid = "mygroup1"

    // 定义kafka相关参数
    val kafkaParams: Map[String, Object] = getKafkaConsumerParameters(groupid)

    //从redis获取offsets
    val offsets: Map[TopicPartition, Long] = OffsetWithRedisUtils.getOffsetsFromRedis(topics,groupid)

    //创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets)
    )
    //DStream转换&输出
    dstream.foreachRDD{(rdd,time)=>
      if(!rdd.isEmpty()){

        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //转换文件内容并发送到topic2
        rdd.foreachPartition{
          iter=>
            iter.map{
              line=>
                line.value().replace("<<<!>>>", "").split(",").mkString("|")
            }.filter(!_.isEmpty).foreach(line=>sendToMytopic2(line,"mytopic2"))
        }
        OffsetWithRedisUtils.saveOffsetsToRedis(offsetRanges,groupid)
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }
}
